package com.nchu.research.research.service.impl;

import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.nchu.research.research.service.DataTransfer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.sql.DataSource;
import java.sql.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author fujianjian
 * @since 2020/12/9 15:30
 */
@Slf4j
@Service
public class DataTransferImpl implements DataTransfer {
    @Autowired
    protected DataSource ihrDataSource;
    @Autowired
    protected DataSource ehrDataSource;

    private static Map<String, Map<Integer, String>> colMaps = new ConcurrentHashMap<>();


    @Override
    public void migrateData(String tableName) {
        Stopwatch stopwatch = Stopwatch.createStarted();
        long offsetId = 0L, pageSize = 1000L;
        long actualSize = 0L;
        try {

            do {
                List<Map<String, Object>> retList = getEHRData(tableName, offsetId, pageSize);
                int size = retList.size();
                log.warn(">>>>migrate table:{}, id offset:{}, page size:{}, actual size:{}",
                        tableName, offsetId, pageSize, actualSize);
                if (CollectionUtils.isEmpty(retList)) {
                    break;
                }
                insertTargetData(tableName, retList);
                actualSize = size;
                Map<Integer, String> tableColMap = initColMap(tableName);
                offsetId  = (long) retList.get(size - 1).get(tableColMap.get(1));
            } while (actualSize == pageSize);
        } finally {
            stopwatch.stop();
            log.warn(">>>>migrate data completed, table:{} offsetId:{}, time cost:{}ms",
                    tableName, offsetId, stopwatch.elapsed(TimeUnit.MILLISECONDS));
        }
    }

    private void insertTargetData(String tableName, List<Map<String, Object>> retList) {
        Stopwatch stopwatch = Stopwatch.createStarted();
        Map<Integer, String> tableColMap = initColMap(tableName);
        StringBuilder sb = new StringBuilder("INSERT IGNORE INTO ");
        sb.append(tableName).append(" (");
        for (String value : tableColMap.values()) {
            sb.append(value).append(",");
        }
        sb.deleteCharAt(sb.length() - 1);
        sb.append(") VALUES (");
        for (Integer key : tableColMap.keySet()) {
            sb.append("?,");
        }
        sb.deleteCharAt(sb.length() - 1);
        sb.append(")");
        try (Connection connection = ihrDataSource.getConnection();) {
            PreparedStatement ps = connection.prepareStatement(sb.toString());
            for (Map<String, Object> tmpMap : retList) {
                for (Integer key : tableColMap.keySet()) {
                    ps.setObject(key, tmpMap.get(tableColMap.get(key)));
                }
                ps.addBatch();
            }
            int[] batchRet = ps.executeBatch();
            log.info("insert target table:{} data, ret:{}", tableName, batchRet);
            //ps.executeLargeBatch();
            ps.close();
        } catch (Exception e) {
            log.error("insert ihr table:{} data error",tableName, e);
        } finally {
            stopwatch.stop();
            log.info("insert ihr table:{} data completed, effect count:{}, time cost:{}ms",
                    tableName, retList.size(), stopwatch.elapsed(TimeUnit.MILLISECONDS));
        }
    }

    private Map<Integer, String> initColMap(String tableName) {
        Map<Integer, String> tableColMap = colMaps.get(tableName);
        if (Objects.isNull(tableColMap)) {
            synchronized (tableName.intern()) {
                tableColMap = colMaps.get(tableName);
                if (Objects.isNull(tableColMap)) {
                    tableColMap = Maps.newConcurrentMap();
                    String colSql = "SELECT ORDINAL_POSITION, COLUMN_NAME from information_schema.`COLUMNS`" +
                            " WHERE TABLE_SCHEMA=? AND TABLE_NAME=? ORDER BY ORDINAL_POSITION";
                    try {
                        Connection connection = ehrDataSource.getConnection();
                        PreparedStatement ps = connection.prepareStatement(colSql);
                        ps.setObject(1, "hr-self-service");
                        ps.setObject(2, tableName);
                        ResultSet resultSet = ps.executeQuery();
                        while (resultSet.next()) {
                            tableColMap.put(resultSet.getInt(1), resultSet.getString(2));
                        }
                        colMaps.put(tableName, tableColMap);
                    } catch (Exception e) {
                        log.info("init table:{} col map error", tableName, e);
                    }
                }
            }
        }
        return tableColMap;
    }

    private List<Map<String, Object>> getEHRData(String tableName, Long offsetId, Long pageSize) {
        Stopwatch stopwatch = Stopwatch.createStarted();
        String fetchSql = String.format("SELECT * from %s WHERE id>? ORDER BY id LIMIT ?", tableName);
        List<Map<String, Object>> retList = null;
        try (Connection connection = ehrDataSource.getConnection()) {
            PreparedStatement ps = connection.prepareStatement(fetchSql);
            ps.setObject(1, offsetId, Types.BIGINT);
            ps.setObject(2, pageSize, Types.BIGINT);
            ResultSet rs = ps.executeQuery();
            retList = Lists.newArrayListWithCapacity(pageSize.intValue());
            while (rs.next()) {
                ResultSetMetaData md = rs.getMetaData();
                int cc = md.getColumnCount();
                Map<String, Object> rowDataMap = Maps.newHashMapWithExpectedSize(cc);
                for (int i = 1; i <= cc; i++) {
                    rowDataMap.put(md.getColumnName(i), rs.getObject(i));
                }
                retList.add(rowDataMap);
            }
            rs.close();
        } catch (Exception e) {
            log.info("get ehr table:{} data error", tableName, e);
        } finally {
            retList = Optional.ofNullable(retList).orElseGet(Collections::emptyList);
            stopwatch.stop();
            log.info("get ehr table:{} data completed, actual size:{}, time cost:{}ms",
                    tableName, retList.size(), stopwatch.elapsed(TimeUnit.MILLISECONDS));
        }
        return retList;
    }
}
